package io.github.talelin.latticy.service.oms.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.lang.Assert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import io.github.talelin.autoconfigure.exception.ForbiddenException;
import io.github.talelin.latticy.bo.InboundOrderBO;
import io.github.talelin.latticy.bo.StockBO;
import io.github.talelin.latticy.common.LocalUser;
import io.github.talelin.latticy.common.enumeration.InboundOrderStatusEnum;
import io.github.talelin.latticy.common.mybatis.Page;
import io.github.talelin.latticy.common.pojo.BillPrefixConfigEntity;
import io.github.talelin.latticy.dto.oms.inbound_order.*;
import io.github.talelin.latticy.mapper.oms.*;
import io.github.talelin.latticy.model.UserDO;
import io.github.talelin.latticy.model.oms.SenderAddressDO;
import io.github.talelin.latticy.model.oms.*;
import io.github.talelin.latticy.service.IdGeneratorService;
import io.github.talelin.latticy.service.SenderAddressService;
import io.github.talelin.latticy.service.mq.StockProducerService;
import io.github.talelin.latticy.service.oms.InboundOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.github.talelin.latticy.service.oms.InboundPackagesItemsService;
import io.github.talelin.latticy.service.oms.InboundPackagesService;
import io.github.talelin.latticy.service.oms.ProductService;
import io.github.talelin.latticy.vo.ReceivingRecordVO;
import io.github.talelin.latticy.vo.oms.InboundOrderDetailsVo;
import io.github.talelin.latticy.vo.oms.InboundOrderPageVo;
import io.seata.spring.annotation.GlobalTransactional;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.function.BiPredicate;
import java.util.stream.Collectors;

/**
 * <p>
 * 入库单表 服务实现类
 * </p>
 *
 * @author generator@mrxiao
 * @since 2022-05-12
 */
@Service
public class InboundOrderServiceImpl extends ServiceImpl<InboundOrderMapper, InboundOrderDO> implements InboundOrderService {

    @Autowired
    private BillPrefixConfigEntity billPrefixConfigEntity;

    @Autowired
    private SenderAddressService senderAddressService;

    @Autowired
    private InboundPackagesService inboundPackagesService;

    @Autowired
    private InboundPackagesItemsService inboundPackagesItemsService;

    @Autowired
    private IdGeneratorService idGeneratorService;

    @Autowired
    private ProductService productService;

    @Autowired
    private StockProducerService stockProducerService;

    @Autowired
    @Qualifier("threadPoolTaskExecutor")
    private Executor executor;

    private BiPredicate<Integer, Integer> isDepositEnough = (totalNumber, receivingTotalCount) -> totalNumber.compareTo(receivingTotalCount) > 0;

    @GlobalTransactional
    @Override
    public void createInboundOrder(InboundOrderCreateDTO createDTO) {
        Optional.ofNullable(createDTO).orElseThrow(() -> new ForbiddenException(19000));
        // 存储过程生成分布式唯一单号
//        String seqNo = tempBillService.getSeqNo(billPrefixConfigEntity.getOrder());
        // redis生成分布式唯一单号
        String seqNo = idGeneratorService.generatorId(billPrefixConfigEntity.getOrder());
        UserDO userDO = LocalUser.getLocalUser();
        InboundOrderDO inboundOrderDO = new InboundOrderDO(createDTO, seqNo, userDO.getUsername());
        // 校验入库单数据，并计算结果
        addInboundOrder(inboundOrderDO, createDTO.getInboundPackagesList());
        this.save(inboundOrderDO);
        // 生成入库单package
        addInboundOrderPackage(createDTO.getSenderAddress(), seqNo, createDTO.getInboundPackagesList());
    }


    @GlobalTransactional
    @Override
    public void updateInboundOrder(InboundOrderUpdateDTO updateVo) {
        Assert.isTrue(!this.checkInboundOrderExistByCode(updateVo.getInboundOrderNo()),"19001");
        InboundOrderDO inboundOrderDO = this.baseMapper.selectById(updateVo.getInboundOrderNo());
        Assert.isTrue(!inboundOrderDO.getInboundOrderStatus().equals(InboundOrderStatusEnum.DRAFT),"19006");
        InboundOrderDO inboundOrder = new InboundOrderDO();
        BeanUtils.copyProperties(updateVo, inboundOrder);
        //封装入库单数据
        addInboundOrder(inboundOrder, updateVo.getInboundPackagesList());
        this.baseMapper.updateById(inboundOrder);
        String inboundOrderNo = updateVo.getInboundOrderNo();
        List<String> inboundOrderNoList = new ArrayList<>();
        inboundOrderNoList.add(inboundOrderNo);
        this.deleteInboundOrder(inboundOrderNoList);
        //封装箱子等相关数据
        addInboundOrderPackage(updateVo.getSenderAddress(), updateVo.getInboundOrderNo(), updateVo.getInboundPackagesList());
    }

    @Override
    public boolean redoInboundOrder(List<ReceivingRecordVO> receivingRecordVOS) {
        receivingRecordVOS.stream().map(receivingRecordVO
                -> CompletableFuture.supplyAsync(() -> reduceAllReceiveNum(receivingRecordVO), executor)).collect(Collectors.toList());
        return true;
    }


    @Override
    public InboundOrderDetailsVo getInboundOrderByNo(String inboundOrderNo) {
        QueryWrapper<InboundOrderDO> inboundOrderWrapper = new QueryWrapper<>();
        inboundOrderWrapper.lambda().eq(InboundOrderDO::getInboundOrderNo, inboundOrderNo);
        InboundOrderDO inboundOrderDO = this.baseMapper.selectOne(inboundOrderWrapper);
        return getInboundOrderInfo(inboundOrderDO);
    }


    @Override
    public boolean checkInboundOrderExistByCode(String inboundOrderNo) {
        QueryWrapper<InboundOrderDO> wrapper = new QueryWrapper<>();
        wrapper.lambda().eq(InboundOrderDO::getInboundOrderNo, inboundOrderNo);
        return this.count(wrapper) > 0;
    }

    @Transactional
    @Override
    public void deleteInboundOrder(List<String> inboundOrderNoList) {
        QueryWrapper<InboundOrderDO> inboundOrderWrapper = new QueryWrapper<>();
        inboundOrderWrapper.lambda().in(InboundOrderDO::getInboundOrderNo, inboundOrderNoList);
        this.baseMapper.delete(inboundOrderWrapper);
        deleteSenderAddressDO(inboundOrderNoList);
        inboundPackagesService.deletePackages(inboundOrderNoList);
    }


    @Transactional
    @Override
    public void draftInboundOrder(InboundOrderListDTO listDto) {
        editInboundOrderStatus(listDto, InboundOrderStatusEnum.CHANCE);
    }

    @GlobalTransactional
    @Override
    public void auditInboundOrder(InboundOrderListDTO listDto) {
        boolean audit = editInboundOrderStatus(listDto, InboundOrderStatusEnum.TO_BE_CONFIRMED);
        if (audit) {
            this.orderBatchDelivery(listDto.getIdList());
        }
    }


    @Override
    public void putAwayInboundOrder(InboundOrderOpenFeignDTO inboundOrderOpenFeignDTO) {
        InboundOrderDO inboundOrder = this.baseMapper.selectById(inboundOrderOpenFeignDTO.getInboundOrderNo());
        Assert.isTrue(inboundOrder.getInboundOrderStatus().equals(InboundOrderStatusEnum.RECEIPT_COMPLETED), "19012");
        InboundOrderDO inboundOrderDO = new InboundOrderDO();
        inboundOrderDO.setInboundOrderNo(inboundOrderOpenFeignDTO.getInboundOrderNo());
        inboundOrderDO.setInboundOrderStatus(InboundOrderStatusEnum.RECEIPT_COMPLETED);
        this.updateById(inboundOrderDO);
    }


    public boolean editInboundOrderStatus(InboundOrderListDTO listDto, InboundOrderStatusEnum statusEnum) {
        List<String> inboundOrderNoList = listDto.getIdList();
        List<InboundOrderDO> inboundOrderDOS = this.baseMapper.selectBatchIds(inboundOrderNoList);
        boolean match = inboundOrderDOS.parallelStream().allMatch(inboundOrderDO -> inboundOrderDO.getInboundOrderStatus().equals(InboundOrderStatusEnum.DRAFT));
        Assert.isTrue(match,"19006");
        List<InboundOrderDO> updateInboundOrderDO = inboundOrderNoList.parallelStream().map(id -> {
            InboundOrderDO inboundOrderDO = new InboundOrderDO();
            inboundOrderDO.setInboundOrderNo(id);
            inboundOrderDO.setInboundOrderStatus(statusEnum);
            return inboundOrderDO;
        }).collect(Collectors.toList());
        return this.updateBatchById(updateInboundOrderDO);
    }

    @Override
    @Transactional
    public void inboundOrderReceiving(ReceivingRecordDTO receivingRecordDTO) {
        if (!this.checkInboundOrderExistByCode(receivingRecordDTO.getInboundOrderNo())) {
            throw new ForbiddenException(19001);
        }
        InboundOrderDO inboundOrderDO = this.baseMapper.selectById(receivingRecordDTO.getInboundOrderNo());
        InboundOrderStatusEnum orderStatus = inboundOrderDO.getInboundOrderStatus();
        if (!orderStatus.equals(InboundOrderStatusEnum.TO_BE_CONFIRMED) && !orderStatus.equals(InboundOrderStatusEnum.RECEIPT)) {
            throw new ForbiddenException(19008);
        }

        List<ReceivingPackageDTO> packages = receivingRecordDTO.getPackages();
        Integer receivingTotalCount = packages.stream()
                // 将packages对象的receiveNum取出来map为Bigdecimal
                .map(m -> m.getReceiveNum())
                // 使用reduce()聚合函数,实现累加器
                .reduce(inboundOrderDO.getReceivingCount(), Integer::sum);

        InboundOrderDO inboundOrderUpdate = new InboundOrderDO();
        inboundOrderUpdate.setInboundOrderNo(receivingRecordDTO.getInboundOrderNo());
        if (isDepositEnough.test(inboundOrderDO.getTotalNumber(), receivingTotalCount)) {
            inboundOrderUpdate.setInboundOrderStatus(InboundOrderStatusEnum.RECEIPT);
        } else {
            inboundOrderUpdate.setInboundOrderStatus(InboundOrderStatusEnum.RECEIPT_COMPLETED);
        }
        inboundOrderUpdate.setReceivingCount(Integer.parseInt(receivingTotalCount.toString()));

        this.baseMapper.updateById(inboundOrderUpdate);

        inboundPackagesService.batchUpdateReceiveQuantityByIds(packages);

        List<CompletableFuture<Boolean>> collect = packages.stream().map(pkg -> CompletableFuture.supplyAsync(
                () -> inboundPackagesItemsService.batchUpdateReceiveQuantityByIds(pkg.getSkus()), executor)).collect(Collectors.toList());
        List<Boolean> resultList = collect.stream().map(CompletableFuture::join).filter(Objects::nonNull).collect(Collectors.toList());
        List<Boolean> errorList = resultList.parallelStream().filter(t -> !t).collect(Collectors.toList());
        Optional.ofNullable(errorList).orElseThrow(() -> new ForbiddenException(19011));
    }

    @Override
    public IPage<InboundOrderPageVo> getInboundOrderPage(InboundOrderPageDTO dto) {
        Page<InboundOrderDO> pager = new Page<>(dto.getCurrPage(), dto.getPageSize());
        Page<InboundOrderDO> inboundOrderDOPage = this.baseMapper.selectPage(pager, buildWrapper(dto));
        List<InboundOrderDO> inboundOrderDOList = inboundOrderDOPage.getRecords();
        Page<InboundOrderPageVo> inboundOrderPageVoPage = new Page<>();
        BeanUtils.copyProperties(inboundOrderDOPage, inboundOrderPageVoPage);
        inboundOrderPageVoPage.setRecords(getInboundOrderDetails(inboundOrderDOList));
        return inboundOrderPageVoPage;
    }

    public void deleteSenderAddressDO(List<String> inboundOrderNoList) {
        QueryWrapper<SenderAddressDO> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().in(SenderAddressDO::getInboundOrderNo, inboundOrderNoList);
        senderAddressService.remove(queryWrapper);
    }

    private InboundOrderDetailsVo getInboundOrderInfo(InboundOrderDO inboundOrderDO) {
        Optional.ofNullable(inboundOrderDO).orElseThrow(() -> new ForbiddenException(19001));
        InboundOrderDetailsVo inboundOrderDetailsVo = new InboundOrderDetailsVo();
        BeanUtils.copyProperties(inboundOrderDO, inboundOrderDetailsVo);
        QueryWrapper<SenderAddressDO> senderAddressDOQueryWrapper = new QueryWrapper<>();
        senderAddressDOQueryWrapper.lambda().eq(SenderAddressDO::getInboundOrderNo, inboundOrderDO.getInboundOrderNo());
        inboundOrderDetailsVo.setSenderAddress(senderAddressService.getOne(senderAddressDOQueryWrapper));
        List<String> inboundOrderList = new ArrayList<>();
        inboundOrderList.add(inboundOrderDO.getInboundOrderNo());
        inboundOrderDetailsVo.setInboundPackagesList(inboundPackagesService.getPackagesMapByOrderNoList(inboundOrderList).get(inboundOrderDO.getInboundOrderNo()));
        return inboundOrderDetailsVo;
    }

    private void addInboundOrder(InboundOrderDO inboundOrderDO, List<InboundPackagesDTO> inboundPackagesList) {

        List<List<Long>> productIdList = inboundPackagesList.parallelStream().map(this::getConvertEntity).map(this::getWarehousePackagesItemList).collect(Collectors.toList());
        Optional.ofNullable(productIdList).orElseThrow(() -> new ForbiddenException(19002));
        List<Long> idList = productIdList.stream().flatMap(Collection::stream).distinct().collect(Collectors.toList());
        Optional.ofNullable(idList).orElseThrow(() -> new ForbiddenException(19003));
        //检查是否包含不存在商品
        this.checkProductIdList(idList);
        //sku总数
        inboundOrderDO.setSkuTotal(idList.size());
        //总箱数
        Integer totalPackages = inboundPackagesList.parallelStream().map(InboundPackagesDTO::getQuantity).reduce(0, Integer::sum);
        inboundOrderDO.setTotalPackages(totalPackages);
        //总件数
        Integer totalNumber = inboundPackagesList.parallelStream().map(InboundPackagesDTO::getInboundPackagesItemsList).map(packagesItems -> packagesItems.parallelStream().
                map(InboundPackagesItemsDTO::getQuantity).reduce(0, Integer::sum)).reduce(0, Integer::sum);
        inboundOrderDO.setTotalNumber(totalNumber);
    }

    private void checkProductIdList(List<Long> idList) {
        int count = productService.checkCountByProductIds(idList);
        if (count != idList.size()) {
            throw new ForbiddenException(19005);
        }
    }

    private List<Long> getWarehousePackagesItemList(InboundPackagesDO inboundPackagesDTO) {
        return inboundPackagesDTO.getInboundPackagesItemsList().parallelStream().map(InboundPackagesItemsDO::getProductId).distinct().collect(Collectors.toList());
    }

    private InboundPackagesDO getConvertEntity(InboundPackagesDTO inboundPackages) {
        InboundPackagesDO inboundPackagesDO=new InboundPackagesDO();
        BeanUtils.copyProperties(inboundPackages,inboundPackagesDO);
        //这一步谁有优化空间
        List<InboundPackagesItemsDO> inboundPackagesItemsList=inboundPackages.getInboundPackagesItemsList().parallelStream().map(items->{
            InboundPackagesItemsDO inboundPackagesItemsDO=new InboundPackagesItemsDO();
            BeanUtils.copyProperties(items,inboundPackagesItemsDO);
            return inboundPackagesItemsDO;
        }).collect(Collectors.toList());
        inboundPackagesDO.setInboundPackagesItemsList(inboundPackagesItemsList);
        return inboundPackagesDO;
    }

    /**
     * 入库单添加封装
     *
     * @param senderAddress
     * @param seqNo
     * @param warehousePackageDOList
     */
    private void addInboundOrderPackage(SenderAddressDO senderAddress, String seqNo, List<InboundPackagesDTO> warehousePackageDOList) {
        senderAddress.setInboundOrderNo(seqNo);
        Optional.ofNullable(senderAddress).ifPresent(this::checkSaveAddress);
        List<CompletableFuture<Boolean>> collect = warehousePackageDOList.stream().map(pkg -> CompletableFuture.supplyAsync(() -> savePackage(pkg, seqNo), executor)).collect(Collectors.toList());
        List<Boolean> resultList = collect.stream().map(CompletableFuture::join).filter(Objects::nonNull).collect(Collectors.toList());
        List<Boolean> errorList = resultList.parallelStream().filter(t -> !t).collect(Collectors.toList());
        Optional.ofNullable(errorList).orElseThrow(() -> new ForbiddenException(19010));
    }

    private void checkSaveAddress(SenderAddressDO senderAddressDO) {
        boolean exist = senderAddressService.save(senderAddressDO);
        if (!exist) {
            throw new ForbiddenException(19009);
        }
    }

    private boolean savePackage(InboundPackagesDTO pkg, String seqNo) {
        pkg.setInboundOrderNo(seqNo);
        return inboundPackagesService.saveInboundPackageDO(pkg);
    }

    private boolean reduceAllReceiveNum(ReceivingRecordVO receivingRecordVO) {
        Assert.isTrue(this.updateReceiveQuantityByPackageNo(receivingRecordVO));
        Assert.isTrue(inboundPackagesService.updateReceiveQuantityByPackageNo(receivingRecordVO));
        Assert.isTrue(inboundPackagesItemsService.updateReceiveQuantityByPackageNo(receivingRecordVO));
        return true;
    }

    private boolean updateReceiveQuantityByPackageNo(ReceivingRecordVO receivingRecordVO) {
        UpdateWrapper updateWrapper = new UpdateWrapper();
        updateWrapper.eq("inbound_order_no", receivingRecordVO.getInboundOrderNo());
        updateWrapper.setSql("receive_quantity = receive_quantity -" + receivingRecordVO.getReceiveNum());
        updateWrapper.setSql("specimen = specimen -" + receivingRecordVO.getSpecimen());
        updateWrapper.setSql("damaged = damaged -" + receivingRecordVO.getDamaged());
        return this.update(updateWrapper);
    }

    private LambdaQueryWrapper buildWrapper(InboundOrderPageDTO dto) {
        QueryWrapper<InboundOrderDO> wrapper = new QueryWrapper<>();
        List<String> idList = new ArrayList<>();
        if (StringUtils.isNotBlank(dto.getSku())) {
            idList = this.baseMapper.selectBySku(dto.getSku());
        }
       List<InboundOrderStatusEnum> inboundOrderStatusList=new ArrayList<>();
        if(dto.getInboundOrderStatus()!=null){
            if(dto.getInboundOrderStatus().equals(InboundOrderStatusEnum.ON_THE_WAY)){
                inboundOrderStatusList.add(InboundOrderStatusEnum.TO_BE_CONFIRMED);
                inboundOrderStatusList.add(InboundOrderStatusEnum.RECEIPT);

            }
        }
        return wrapper.lambda()
                .eq(dto.getInboundOrderStatus() != null&&CollectionUtils.isEmpty(inboundOrderStatusList), InboundOrderDO::getInboundOrderStatus, dto.getInboundOrderStatus())
                .in(CollectionUtils.isNotEmpty(inboundOrderStatusList), InboundOrderDO::getInboundOrderStatus, inboundOrderStatusList)
                .eq(dto.getInboundOrderType() != null, InboundOrderDO::getInboundOrderType, dto.getInboundOrderType())
                .eq(dto.getContainerType() != null, InboundOrderDO::getContainerType, dto.getContainerType())
                .in(CollectionUtils.isNotEmpty(idList), InboundOrderDO::getInboundOrderNo, idList)
                .between(dto.getBeginTime() != null && dto.getEndTime() != null, InboundOrderDO::getCreateTime, dto.getBeginTime(), dto.getEndTime())
                .like(StringUtils.isNotBlank(dto.getInboundOrderNo()), InboundOrderDO::getInboundOrderNo, dto.getInboundOrderNo())
                .in(CollUtil.isNotEmpty(dto.getWarehouseCodes()), InboundOrderDO::getWarehouseCode, dto.getWarehouseCodes())
                .eq(StringUtils.isNotEmpty(dto.getTenant()), InboundOrderDO::getTenant, dto.getTenant())
                .like(StringUtils.isNotBlank(dto.getPurchaseOrderNo()), InboundOrderDO::getPurchaseOrderNo, dto.getPurchaseOrderNo())
                .like(StringUtils.isNotBlank(dto.getTrackingNumber()), InboundOrderDO::getTrackingNumber, dto.getTrackingNumber());
    }


    public List<InboundOrderPageVo> getInboundOrderDetails(List<InboundOrderDO> inboundOrderDOs) {
        List<InboundOrderPageVo> inboundOrderPageVoList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(inboundOrderDOs)) {
            List<String> inboundOrderNoList = inboundOrderDOs.parallelStream().map(InboundOrderDO::getInboundOrderNo).collect(Collectors.toList());
            Map<String, List<InboundPackagesDO>> packagesMap = inboundPackagesService.getPackagesMapByOrderNoList(inboundOrderNoList);

            inboundOrderDOs.forEach(inboundOrderDO -> {
                InboundOrderPageVo inboundOrderPageVo = new InboundOrderPageVo();
                BeanUtils.copyProperties(inboundOrderDO, inboundOrderPageVo);
                List<InboundPackagesDO> inboundPackagesDOS = packagesMap.get(inboundOrderDO.getInboundOrderNo());
                if(CollectionUtils.isNotEmpty(inboundPackagesDOS)) {
                    List<InboundItemCountDTO> inboundItemCount = inboundPackagesDOS.parallelStream().map(InboundPackagesDO::getInboundPackagesItemsList).collect(Collectors.toList()).parallelStream()
                            .flatMap(Collection::stream).distinct().collect(Collectors.toList()).
                                    parallelStream().collect(Collectors.groupingBy(InboundPackagesItemsDO::getSku)).entrySet().parallelStream().map(packagesItems -> {
                                InboundItemCountDTO inboundItemCountDTO = new InboundItemCountDTO();
                                Integer quantity = packagesItems.getValue().parallelStream().map(InboundPackagesItemsDO::getQuantity).reduce(0, Integer::sum);
                                inboundItemCountDTO.setSku(packagesItems.getKey());
                                inboundItemCountDTO.setQuantity(quantity);
                                return inboundItemCountDTO;
                            }).collect(Collectors.toList());
                    inboundOrderPageVo.setChildren(inboundItemCount);
                    inboundOrderPageVoList.add(inboundOrderPageVo);
                }
            });
        }
        return inboundOrderPageVoList;
    }

    private List<StockBO> generateStockBOS(String warehouseCode, List<InboundPackagesDO> inboundPackagesList) {
        // 获取一个入库单下的所有产品id列表
        List<List<Long>> productIdList = inboundPackagesList.parallelStream().
                map(this::getWarehousePackagesItemList).collect(Collectors.toList());
        Optional.ofNullable(productIdList).orElseThrow(() -> new ForbiddenException(19002));
        List<Long> idList = productIdList.stream().flatMap(Collection::stream).distinct().collect(Collectors.toList());
        Optional.ofNullable(idList).orElseThrow(() -> new ForbiddenException(19003));

        // 通过产品id列表获取产品列表
        List<ProductDO> productDOList = productService.getBaseMapper().selectBatchIds(idList);
        // 生成产品map
        Map<Long, ProductDO> productDOMap = productDOList.parallelStream().collect(Collectors.toMap(ProductDO::getId, productDO -> productDO));
        UserDO userDO = LocalUser.getLocalUser();
        String tenant = userDO.getTenant();
        List<CompletableFuture<List<StockBO>>> collect = inboundPackagesList.stream().map(pkg -> CompletableFuture.supplyAsync(() -> generateStockBO(warehouseCode, tenant, pkg, productDOMap), executor)).collect(Collectors.toList());
        List<List<StockBO>> stockBOList = collect.stream().map(CompletableFuture::join).filter(Objects::nonNull).collect(Collectors.toList());
        List<StockBO> allStocks = stockBOList.stream().flatMap(Collection::stream).collect(Collectors.toList());
        List<StockBO> lastStockList = new ArrayList<>();
        allStocks.parallelStream().collect(Collectors.groupingBy(s -> (s.getSku()), Collectors.toList())).forEach((id, transfer) -> {
            transfer.stream().reduce((a, b) -> new StockBO(a.getWarehouseCode(), tenant, a.getSku(), a.getSkuName(),
                    a.getOnTheWay() + b.getOnTheWay(), a.getProductTime(),new Date())).ifPresent(lastStockList::add);
        });
        return lastStockList;
    }

    private List<StockBO> generateStockBO(String warehouseCode, String tenant, InboundPackagesDO inboundPackagesDTO, Map<Long, ProductDO> productDOMap) {
        return inboundPackagesDTO.getInboundPackagesItemsList().parallelStream().map(
                pkgItem -> StockBO.builder().warehouseCode(warehouseCode).tenant(tenant)
                        .sku(productDOMap.get(pkgItem.getProductId()).getSku())
                        .skuName(productDOMap.get(pkgItem.getProductId()).getName())
                        .onTheWay(Long.valueOf(pkgItem.getQuantity()))
                        .productTime(new Date())
                        .build()).distinct().collect(Collectors.toList());
    }


    private void orderBatchDelivery(List<String> inboundOrderNoList) {
        List<InboundPackagesDO> inboundPackagesDOS = inboundPackagesService.getPackagesByOrderNoList(inboundOrderNoList);
        if (CollUtil.isEmpty(inboundPackagesDOS)) {
            throw new ForbiddenException(19002);
        }
        // 查询入库单详情
        Map<String, InboundOrderBO> orderDetailMap = this.baseMapper.getInboundOrderBOListByIds(inboundOrderNoList);
        Map<String, List<InboundPackagesDO>> packagesMap = inboundPackagesDOS.parallelStream().collect(Collectors.groupingBy(InboundPackagesDO::getInboundOrderNo));
        List<String> packagesNoList = inboundPackagesDOS.parallelStream().map(InboundPackagesDO::getPackageNo).collect(Collectors.toList());
        Map<String, List<InboundPackagesItemsDO>> packagesItemsMap = inboundPackagesItemsService.getPackagesItemsMapByPackageNoList(packagesNoList);
        inboundOrderNoList.forEach(inboundOrderNo ->generateStockBOSByInboundOrderNo(inboundOrderNo, packagesMap, packagesItemsMap, orderDetailMap));
    }

    private Boolean generateStockBOSByInboundOrderNo(String inboundOrderNo, Map<String, List<InboundPackagesDO>> packagesMap, Map<String, List<InboundPackagesItemsDO>> packagesItemsMap, Map<String, InboundOrderBO> orderDetailMap) {
        List<InboundPackagesDO> inboundPackagesList = packagesMap.get(inboundOrderNo);
        List<CompletableFuture<List<StockBO>>> collect = inboundPackagesList.stream()
                .map(pkg -> CompletableFuture.supplyAsync(() -> generateStockBOByInboundOrderNo(orderDetailMap.get(inboundOrderNo).getWarehouseCode(), orderDetailMap.get(inboundOrderNo).getTenant(), pkg, packagesItemsMap), executor)).collect(Collectors.toList());
        List<List<StockBO>> stockBOList = collect.stream().map(CompletableFuture::join).filter(Objects::nonNull).collect(Collectors.toList());
        List<StockBO> allStocks = stockBOList.stream().flatMap(Collection::stream).collect(Collectors.toList());
        List<StockBO> lastStockList = new ArrayList<>();
        allStocks.parallelStream().collect(Collectors.groupingBy(StockBO::getSku, Collectors.toList())).forEach((id, transfer) -> {
            transfer.stream().reduce((a, b) -> new StockBO(a.getWarehouseCode(), a.getTenant(), a.getSku(), a.getSkuName(),
                    a.getOnTheWay() + b.getOnTheWay(), a.getProductTime(),new Date())).ifPresent(lastStockList::add);
        });
        // 投递在途库存消息
        stockProducerService.deliveryOnTheWayStock(lastStockList);
        return true;
    }

    private List<StockBO> generateStockBOByInboundOrderNo(String warehouseCode, String tenant, InboundPackagesDO inboundPackagesDO, Map<String, List<InboundPackagesItemsDO>> packagesItemsMap) {
        return packagesItemsMap.get(inboundPackagesDO.getPackageNo()).parallelStream().map(
                pkgItem -> StockBO.builder().warehouseCode(warehouseCode).tenant(tenant)
                        .sku(pkgItem.getSku())
                        .tenant(tenant)
                        .skuName(pkgItem.getSkuName())
                        .onTheWay(Long.valueOf(pkgItem.getQuantity()))
                        .productTime(new Date())
                        .build()).distinct().collect(Collectors.toList());
    }

}
